package mc.support.zmq.mla;

import mc.support.proto.Biz;
import mc.support.proto.GeneralReq;
import mc.support.proto.GeneralRsp;
import mc.support.task.common.CronTaskAdaptor;
import mc.support.task.common.Waitable;
import mc.support.util.PlainUtil;
import mc.support.zmq.ProtoConstants;
import mc.support.zmq.ProtoHelper;
import mc.support.zmq.err.ProtoException;
import mc.support.zmq.route.ZmqServerRegister;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;

import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created with IntelliJ IDEA. <br/>
 * Author: Francis Yuen    <br/>
 * Date: 2014-10-28  <br/>
 */
public class MLAProtoServer extends CronTaskAdaptor implements MLAServices, MLAConstants, ProtoConstants {

    private static final Logger log = LoggerFactory.getLogger( MLAProtoServer.class );

    private static final AtomicInteger poolNumber = new AtomicInteger( 1 );
    //TODO optimize to be done!
    private static final ExecutorService exec = Executors.newCachedThreadPool(
            new ThreadFactory() {
                private final ThreadGroup group;
                private final AtomicInteger threadNumber = new AtomicInteger( 1 );
                private final String namePrefix;

                {
                    SecurityManager s = System.getSecurityManager();
                    group = ( s != null ) ? s.getThreadGroup() :
                            Thread.currentThread().getThreadGroup();
                    namePrefix = "mla-worker-pool-" +
                            poolNumber.getAndIncrement() +
                            "-thread-";
                }

                @Override
                public Thread newThread( Runnable r ) {
                    Thread t = new Thread( group, r,
                            namePrefix + threadNumber.getAndIncrement(),
                            0 );

                    t.setDaemon( true );
                    if ( t.getPriority() != Thread.NORM_PRIORITY ) {
                        t.setPriority( Thread.NORM_PRIORITY );
                    }
                    t.setUncaughtExceptionHandler( new Thread.UncaughtExceptionHandler() {
                        @Override
                        public void uncaughtException( Thread t, Throwable e ) {
                            log.error( "Uncaught Exception Detected: {} in Thread: {} ",
                                    new Object[] {PlainUtil.getStackTrace( e ), t.getName()} );
                        }
                    } );
                    return t;
                }

            }
    );

    private static final String IPv4_PAT =
            "^([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\." +
                    "([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\." +
                    "([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\." +
                    "([01]?\\d\\d?|2[0-4]\\d|25[0-5])$";

    public static String getLocalIpByInterfaceName( String name ) {
        if ( null == name ) return null;
        String last = null;
        try {
            Enumeration e = NetworkInterface.getNetworkInterfaces();
            while ( e.hasMoreElements() ) {
                NetworkInterface n = (NetworkInterface) e.nextElement();

                Enumeration ee = n.getInetAddresses();
                while ( ee.hasMoreElements() ) {
                    InetAddress i = (InetAddress) ee.nextElement();
                    last = i.getHostAddress();
                    if ( name.equals( n.getDisplayName() ) && last.matches( IPv4_PAT ) ) {
                        return last;
                    }
                }
            }
        } catch ( SocketException err ) {
            log.error( "get local ip error: {} ", PlainUtil.getStackTrace( err ) );
        }
        return last;
    }

    private int ioThreads;
    private int workerNumber;
    private String address;
    private String serviceInterface;
    private int port;
    private ZmqServerRegister zmqServerRegister;
    private Waitable waitable;
    private final Map<Biz, BizHandler<GeneralReq, GeneralRsp>> handlers;
    private final BizHandler<GeneralReq, GeneralRsp> handlerWrapper;

    {
        handlers = new HashMap<>();
        handlerWrapper = new BizHandler<GeneralReq, GeneralRsp>() {
            @Override
            public GeneralRsp handle( GeneralReq req ) {
                Biz biz = req.getBiz();
                if ( MAGIC_CODE != req.getMagicCode() ) {
                    throw new ProtoException( "magic code not match!", ProtoException.INVALID_PROTO_VERSION );
                }
                if ( Biz.SYS_PING == biz ) {
                    return ProtoHelper.sysPingRsp();
                }
                BizHandler<GeneralReq, GeneralRsp> realHandler = handlers.get( biz );
                if ( null == realHandler ) {
                    throw new ProtoException( "no available biz handler!", ProtoException.NO_AVAILABLE_HANDLER );
                }
                return realHandler.handle( req );
            }
        };
    }

    public void setIoThreads( int ioThreads ) {
        this.ioThreads = ioThreads;
    }

    public void setWorkerNumber( int workerNumber ) {
        this.workerNumber = workerNumber;
    }

    public void setServiceInterface( String serviceInterface ) {
        this.serviceInterface = serviceInterface;
    }

    public void setAddress( String address ) {
        this.address = address;
    }

    public void setWaitable( Waitable waitable ) {
        this.waitable = waitable;
    }

    public String getAddress() {
        if ( null == address ) {
            String ip = getLocalIpByInterfaceName( serviceInterface );
            if ( null != ip ) {
                address = "tcp://" + ip + ":" + port;
            }
        }
        return address;
    }

    public void setPort( int port ) {
        this.port = port;
    }

    public void setZmqServerRegister( ZmqServerRegister zmqServerRegister ) {
        this.zmqServerRegister = zmqServerRegister;
    }

    private ZMQ.Socket clients;
    private ZMQ.Socket workers;
    private ZMQ.Context context;
    private AtomicBoolean workerKeepOn = new AtomicBoolean( true );

    @Override
    public void run() {
        if ( null != waitable ) waitable.waiting();
        if ( 0 == ioThreads ) ioThreads = 1;
        context = ZMQ.context( ioThreads );
        clients = context.socket( ZMQ.ROUTER );
        clients.bind( "tcp://*:" + port );
        workers = context.socket( ZMQ.DEALER );
        workers.bind( "inproc://" + MLA_WORKERS );
        launchWorkers( context );
        zmqServerRegister.register( Biz.SYS_STUB.toString(), getAddress() );
        zmqServerRegister.register( Biz.SYS_PING.toString(), getAddress() );
        ZMQ.proxy( clients, workers, null ); //blocking here
    }

    @Override
    public void setKeepOn( boolean keepOn ) {
        if ( !keepOn ) {
            workerKeepOn.set( false );
            if ( null != clients ) clients.close();
            if ( null != workers ) workers.close();
//            if ( null != context ) context.term();
        }
        super.setKeepOn( keepOn );
    }

    private void launchWorkers( final ZMQ.Context context ) {
        final CountDownLatch latch = new CountDownLatch( workerNumber );
        final AtomicInteger insCount = new AtomicInteger( 0 );
        for ( int i = 0; i < workerNumber; i++ ) {
            exec.execute( newWorker( context, latch, insCount ) );
        }
        try {
            latch.await();
        } catch ( InterruptedException ignored ) {
        }
    }

    private MLAWorker newWorker( final ZMQ.Context context, final CountDownLatch latch, final AtomicInteger insCount ) {
        return new MLAWorker() {
            @Override
            void prepared() {
                log.info( "MLA worker instance {} launched!", insCount.getAndIncrement() );
                latch.countDown();
            }
        }.setKeepOn( workerKeepOn )
                .setContext( context )
                .setGeneralHandler( handlerWrapper );
    }

    @Override
    public void register( Biz biz, BizHandler<GeneralReq, GeneralRsp> handler ) {
        synchronized ( handlers ) {
            if ( null != biz && null != handler ) {
                log.info( "Handler of Biz: {} registered!", biz.name() );
                String address = getAddress();
                if ( null != address ) {
                    log.info( "publishing service address {} ", address );
                    zmqServerRegister.register( biz.toString(), address );
                }
                handlers.put( biz, handler );
            }
        }
    }

    @Override
    public boolean isLaunchNow() {
        return true;
    }

}
